package com.travel.programApp

import java.util

import com.travel.common.{Constants, District}
import com.travel.utils.{HbaseTools, SparkUtils}
import com.uber.h3core.H3Core
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.log4j.{Level, Logger}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.geotools.geometry.jts.JTSFactoryFinder
import org.locationtech.jts.geom.{GeometryFactory, Point, Polygon}
import org.locationtech.jts.io.WKTReader

import scala.collection.mutable

object SparkSQLVirtualStation {

  /**
    * 虚拟车站的计算是实时的计算还是离线的计算？？？
    *    使用离线的计算，将计算的结果，保存到hbase里面去
    */
  private val h3: H3Core = H3Core.newInstance()

  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.ERROR)

    val conf = new SparkConf()
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.setMaster("local[1]").setAppName("sparkHbase")
    val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    val context: SparkContext = sparkSession.sparkContext
    context.setLogLevel("WARN")
    //hbase配置
    val hconf: Configuration = HBaseConfiguration.create()
    hconf.set("hbase.zookeeper.quorum", "node01,node02,node03")
    hconf.set("hbase.zookeeper.property.clientPort", "2181")
    hconf.setInt("hbase.client.operation.timeout", 3000)
    val hbaseFrame: DataFrame = HbaseTools.loadHBaseData(sparkSession,hconf)
    //将数据注册成为一张表，通过sql的方式去操作数据
    hbaseFrame.createOrReplaceTempView("order_df")
    ///通过在已定义函数来实现计算每一个六边形的上车的点
   val virtual_rdd: RDD[Row] = SparkUtils.getVirtualFrame(sparkSession)  //上车点已经有了
    //每一个上车点属于哪一个区如何计算？？？
    //每一个区都有一个边界  一圈边界是通过经纬度来表示的
    //如何获取每一个区的边界    判断上车点属于哪一个区
    //上车点有很多 数据量比较大  边界数据比较少的一份数据  ==》 做广播变量

    //调用高德地图或者腾讯地图  传入一个区的名字，给我返回一个边界范围值
    //context.broadcast(//边界数据广播出去)
    //所有的广播的边界都在这里
    val districtBroadCastVar: Broadcast[util.ArrayList[District]] = SparkUtils.broadCastDistrictValue(sparkSession)
    //边界已经获取到了   计算每一个上车点与边界的关系
    //通过java geometry  来判断  区域与点的关系  包含还是不包含
    val finalSaveRow: RDD[mutable.Buffer[Row]] = virtual_rdd.mapPartitions(eachPartition => {
      //将每一个地区，转换成为java Geometry对象
      val geometryFactory: GeometryFactory = JTSFactoryFinder.getGeometryFactory(null)
      //通过geometryFactory来画多边形
      val reader = new WKTReader(geometryFactory)

      //将广播出去每一个地区的形状，转换成为元组  每一个地区，对应一圈边界的经纬度坐标
      val wktPolygons: mutable.Buffer[(District, Polygon)] = SparkUtils.changeDistictToPolygon(districtBroadCastVar, reader)

      //获取到了每一个点
      eachPartition.map(row => {
        //通过每一个点与边界进行比较
        val lng: String = row.getAs[String]("starting_lng")
        val lat: String = row.getAs[String]("starting_lat")
        //将经纬度转换成为一个point
        val wktpoint: String = "POINT(" + lng + " " + lat + ")"
        // POINT(lng lat)
        val point: Point = reader.read(wktpoint).asInstanceOf[Point]
        //遍历每一个地区的边界
        val rows: mutable.Buffer[Row] = wktPolygons.map(polygon => {
          if (polygon._2.contains(point)) {
            val fields: Array[Any] = row.toSeq.toArray ++ Seq(polygon._1.getName)
            Row.fromSeq(fields)
          } else {
            null
          }
        }).filter(null != _)
        rows
      })

    })
    val rowRDD: RDD[Row] = finalSaveRow.flatMap(x => x )
    //将计算的结果，保存到hbase里面去   //将rdd的数据写入到hbase的  VIRTUAL_STATIONS  这个表里面去
    HbaseTools.saveOrWriteData(hconf,rowRDD,Constants.VIRTUAL_STATION)

  }

}
